package cn.panshi.mongocdc.redis;

import org.apache.commons.lang.StringUtils;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;

import lombok.extern.slf4j.Slf4j;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisCluster;
import redis.clients.jedis.commands.JedisCommands;
import redis.clients.jedis.params.SetParams;

@Slf4j
@SuppressWarnings({ "unchecked", "rawtypes" })
public abstract class AbstractRedisLockTemplate {
	private RedisTemplate redisTemplate;
	private long expireSeconds;
	private String lockToken;
	private boolean unlockAfterSuccess = true;

	/**
	 * @param lockToken 解锁时用于验证身份——解铃还须系铃人
	 * @param unlockAfterSuccess 加锁并完成操作后，是否主动释放锁：
	 * --有些操作过程中并没有幂等、防重检查，如果在完成后主动解锁 可能导致重复;
	 * --如果在doBizOnLockSuccess过程中有幂等、防重检查，则可以在操作完成后主动解锁
	 */
	public AbstractRedisLockTemplate(RedisTemplate redisTemplate, long expireSeconds, String lockToken,
			boolean unlockAfterSuccess) {
		super();
		this.redisTemplate = redisTemplate;
		this.expireSeconds = expireSeconds;
		this.lockToken = lockToken;
		this.unlockAfterSuccess = unlockAfterSuccess;
	}

	public final void doBizWithinLockProtection(IRedisNxLockProtected rlp) {
		String key = constructKey();

		boolean lockSuccess = this.lock(key);
		try {
			if (!lockSuccess) {
				//log.debug("############lock failed for key {}", key);
				rlp.onLockFailure(key);
			} else {
				log.debug("############lock success for key {}", key);
				rlp.doBizOnLockSuccess(key);
			}
		} finally {
			if (lockSuccess && unlockAfterSuccess) {
				Long unlock = this.unlock(key);

				log.debug("release lock success for key {} token {}", key, lockToken);

				rlp.afterReleaseLock(unlock);
			}
		}
	}

	private boolean lock(String key) {

		final String result = (String) redisTemplate.execute(new RedisCallback<String>() {
			@Override
			public String doInRedis(RedisConnection connection) throws DataAccessException {
				JedisCommands commands = (JedisCommands) connection.getNativeConnection();

				SetParams params = new SetParams().nx().ex(expireSeconds);
				return commands.set("LOCK:" + key, lockToken, params);
			}
		});

		log.trace("lock {} for key {}", result, key);
		boolean lockSuccess = (StringUtils.equals(result, "OK"));

		return lockSuccess;
	}

	protected abstract String constructKey();

	private Long unlock(String key) {
		String UNLOCK_LUA = "if redis.call(\"get\",KEYS[1]) == ARGV[1]                                      "
				+ "              then  return redis.call(\"del\",KEYS[1])\r\n                               "
				+ "                else  return 0                                                           "
				+ "          end                                                                            ";

		Long result = (Long) redisTemplate.execute(new RedisCallback<Long>() {
			@Override
			public Long doInRedis(RedisConnection connection) throws DataAccessException {
				Object nativeConnection = connection.getNativeConnection();
				if (nativeConnection instanceof JedisCluster) {
					log.debug(lockToken);
					return (Long) ((JedisCluster) nativeConnection).eval(UNLOCK_LUA, 1, "LOCK:" + key, lockToken);
				}

				else if (nativeConnection instanceof Jedis) {
					log.trace("unlock by lockToken " + lockToken);
					return (Long) ((Jedis) nativeConnection).eval(UNLOCK_LUA, 1, "LOCK:" + key, lockToken);
				}
				return 0L;
			}
		});

		return result;
	}
}